package trouger.dmt;

import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Iterator;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.ReflectionUtils;

class BaseReducer <T extends ReducerEntity<K, V>, K, V>
extends MapReduceBase
implements Reducer<K, V, NullWritable, Text>{
	
	protected int sourceCount;
	protected int maxReduceItem = 0;
	protected String separator;
	protected String nullValueNotation;
	protected int resultColumnCount;
	protected int[][] keyColumnList;
	protected int[][] valueColumnList;
	protected FieldIndexPair[] resultFieldsIndex;
	
	protected T reducerEntity;
	RecordGroupFilter recordGroupFilter;
	OutputRecordFilter outputRecordFilter;
	
	@SuppressWarnings("unchecked")
	@Override
	public void configure(JobConf conf){
		//source-count
		sourceCount = conf.getInt(BaseMapredParams.SOURCE_COUNT_PARAM_NAME, 0);
		maxReduceItem = conf.getInt(BaseMapredParams.MAX_REDUCE_ITEM_COUNT_PARAM_NAME, 0);
		if (sourceCount <= 0){
			throw new IllegalArgumentException("a positive source-count parameter should be provided");
		}
		//separator
		separator = BaseMapredUtils.getConfigString(conf, BaseMapredParams.SEPARATOR_PARAM_NAME);
		if (separator == null){
			separator = "\t";
		}
		//null-value notation
		nullValueNotation = BaseMapredUtils.getConfigString(conf, BaseMapredParams.NULLNOTATION_PARAM_NAME);
		if (nullValueNotation == null){
			nullValueNotation = "null";
		}
		//entity class
		Class<T> reducerEntityClass;
		Type c = ((ParameterizedType)this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
		reducerEntityClass = (Class<T>)c;
		reducerEntity = ReflectionUtils.newInstance(reducerEntityClass, conf);
		//key/value column list
		this.keyColumnList = BaseMapredUtils.getKeyColumnList(sourceCount, conf);
		this.valueColumnList = BaseMapredUtils.getValueColumnListEx(sourceCount, conf, keyColumnList);
		//record group filter
		String recordGroupFilterClassName = conf.get(BaseMapredParams.RECORDGROUPFILTER_CLASSNAME_PARAM_NAME);
		if (recordGroupFilterClassName != null){
			try {
				RecordGroupFilter filter = ReflectionUtils.newInstance( 
					(Class<RecordGroupFilter>) Class.forName(recordGroupFilterClassName),
					conf);
				this.recordGroupFilter = filter;
			} catch (ClassNotFoundException e) {
				throw new IllegalArgumentException("class not found: " + recordGroupFilterClassName);
			}
		}
		//output record filter
		String outputRecordFilterClassName = conf.get(BaseMapredParams.OUTPUTRECORDFILTER_CLASSNAME_PARAM_NAME);
		if (outputRecordFilterClassName != null){
			try {
				this.outputRecordFilter = ReflectionUtils.newInstance(
						(Class<OutputRecordFilter>)Class.forName(outputRecordFilterClassName),
						conf);
			} catch (ClassNotFoundException e) {
				throw new IllegalArgumentException("class not found: " + outputRecordFilterClassName);
			}
		}
		//result fields
		if (recordGroupFilter == null){
			this.resultFieldsIndex = BaseMapredUtils.getResultFieldIndices(conf, sourceCount);
			if (this.resultFieldsIndex == null){
				throw new IllegalArgumentException("record-group-filter and result-fields-indices cannot be both empty");
			}
		}
	}
	
	protected class RecordOutputerImpl implements RecordOutputer{
		protected OutputCollector<NullWritable, Text> outputCollector;
		protected StringBuffer strbuf;
		protected Text outValue = new Text();
		protected OutputRecordFilter filter = null;
		
		public RecordOutputerImpl(OutputCollector<NullWritable, Text> output) {
			this.outputCollector = output;
			this.strbuf = new StringBuffer();
		}
		@Override
		public void output(Object[] record) {
			int n = record.length;
			if (resultColumnCount > 0 && n - 1 != resultColumnCount){
				throw new IllegalArgumentException("the number of record-columns is incorrect");
			}
			if (filter != null){
				if (filter.filterRecord(record) == false){
					return;
				}
			}
			for (int i = 1; i < n; i++){
				if (i > 1) strbuf.append(separator);
				if (record[i] != null){
					strbuf.append(record[i].toString());
				}else{
					strbuf.append(nullValueNotation);
				}
			}
			outValue.set(strbuf.toString());
			strbuf.delete(0, strbuf.length());
			try {
				outputCollector.collect(null, outValue);
			} catch (IOException e) {
				System.err.println(e.toString());
			}
		}
		@Override
		public OutputRecordFilter getOutputRecordFilter() {
			return filter;
		}
		@Override
		public void setOutputRecordFilter(OutputRecordFilter filter) {
			this.filter = filter;
		}
	}
	
	@Override
	public void reduce(K key, Iterator<V> values,
			OutputCollector<NullWritable, Text> output, Reporter reporter)
			throws IOException {
		RecordOutputer outputer = new RecordOutputerImpl(output);
		if (outputRecordFilter != null){
			outputer.setOutputRecordFilter(outputRecordFilter);
		}
		reducerEntity.acceptRecordGroup(key, values, outputer);
	}

}
